use std::{collections::HashMap, ffi::{c_int, CString}, hash::BuildHasherDefault, io, mem, ops::RangeFrom, ptr, thread};

use fnv::FnvHasher;
use mio::{unix::SourceFd, Events, Interest, Poll, Token};

use super::{error::UnixError,
    ipc_receiver::{OsIpcReceiver, PollEntry},
    ipc_sender::OsIpcSender,
    shared_memory::OsIpcSharedMemory,
    utils,
    message::{self, BlockingMode},
};




pub fn channel() -> Result<(OsIpcSender, OsIpcReceiver), UnixError> {
    let mut results = [0, 0];
    unsafe {
        if libc::socketpair(
            libc::AF_UNIX,
            libc::SOCK_SEQPACKET | utils::SOCK_FLAGS,
            0,
            &mut results[0],
        ) >= 0
        {
            Ok((
                OsIpcSender::from_fd(results[0]),
                OsIpcReceiver::from_fd(results[1]),
            ))
        } else {
            Err(UnixError::last())
        }
    }
}



#[derive(PartialEq, Debug)]
pub enum OsIpcChannel {
    Sender(OsIpcSender),
    Receiver(OsIpcReceiver),
}

impl OsIpcChannel {
    pub(super) fn fd(&self) -> c_int {
        match *self {
            OsIpcChannel::Sender(ref sender) => sender.fd.0,
            OsIpcChannel::Receiver(ref receiver) => receiver.fd.get(),
        }
    }
}

pub struct OsIpcReceiverSet {
    incrementor: RangeFrom<u64>,
    poll: Poll,
    pollfds: HashMap<Token, PollEntry, BuildHasherDefault<FnvHasher>>,
    events: Events,
}

impl Drop for OsIpcReceiverSet {
    fn drop(&mut self) {
        for &PollEntry { id: _, fd } in self.pollfds.values() {
            let result = unsafe { libc::close(fd) };
            assert!(thread::panicking() || result == 0);
        }
    }
}

impl OsIpcReceiverSet {
    pub fn new() -> Result<OsIpcReceiverSet, UnixError> {
        let fnv = BuildHasherDefault::<FnvHasher>::default();
        Ok(OsIpcReceiverSet {
            incrementor: 0..,
            poll: Poll::new()?,
            pollfds: HashMap::with_hasher(fnv),
            events: Events::with_capacity(10),
        })
    }

    pub fn add(&mut self, receiver: OsIpcReceiver) -> Result<u64, UnixError> {
        let last_index = self.incrementor.next().unwrap();
        let fd = receiver.consume_fd();
        let fd_token = Token(fd as usize);
        let poll_entry = PollEntry { id: last_index, fd };
        self.poll
            .registry()
            .register(&mut SourceFd(&fd), fd_token, Interest::READABLE)?;
        self.pollfds.insert(fd_token, poll_entry);
        Ok(last_index)
    }

    pub fn select(&mut self) -> Result<Vec<OsIpcSelectionResult>, UnixError> {
        // Poll until we receive at least one event.
        loop {
            match self.poll.poll(&mut self.events, None) {
                Ok(()) if !self.events.is_empty() => break,
                Ok(()) => {},
                Err(ref error) => {
                    if error.kind() != io::ErrorKind::Interrupted {
                        return Err(UnixError::last());
                    }
                },
            }

            if !self.events.is_empty() {
                break;
            }
        }

        let mut selection_results = Vec::new();
        for event in self.events.iter() {
            // We only register this `Poll` for readable events.
            assert!(event.is_readable());

            let event_token = event.token();
            let poll_entry = self
                .pollfds
                .get(&event_token)
                .expect("Got event for unknown token.")
                .clone();
            loop {
                match message::recv(poll_entry.fd, BlockingMode::Nonblocking) {
                    Ok((data, channels, shared_memory_regions)) => {
                        selection_results.push(OsIpcSelectionResult::DataReceived(
                            poll_entry.id,
                            data,
                            channels,
                            shared_memory_regions,
                        ));
                    },
                    Err(err) if err.channel_is_closed() => {
                        self.pollfds.remove(&event_token).unwrap();
                        self.poll
                            .registry()
                            .deregister(&mut SourceFd(&poll_entry.fd))
                            .unwrap();
                        unsafe {
                            libc::close(poll_entry.fd);
                        }

                        selection_results.push(OsIpcSelectionResult::ChannelClosed(poll_entry.id));
                        break;
                    },
                    Err(UnixError::Errno(code)) if code == libc::EWOULDBLOCK => {
                        // We tried to read another message from the file descriptor and
                        // it would have blocked, so we have exhausted all of the data
                        // pending to read.
                        break;
                    },
                    Err(err) => return Err(err),
                }
            }
        }

        Ok(selection_results)
    }
}

pub enum OsIpcSelectionResult {
    DataReceived(
        u64,
        Vec<u8>,
        Vec<OsOpaqueIpcChannel>,
        Vec<OsIpcSharedMemory>,
    ),
    ChannelClosed(u64),
}

impl OsIpcSelectionResult {
    pub fn unwrap(
        self,
    ) -> (
        u64,
        Vec<u8>,
        Vec<OsOpaqueIpcChannel>,
        Vec<OsIpcSharedMemory>,
    ) {
        match self {
            OsIpcSelectionResult::DataReceived(id, data, channels, shared_memory_regions) => {
                (id, data, channels, shared_memory_regions)
            },
            OsIpcSelectionResult::ChannelClosed(id) => {
                panic!(
                    "OsIpcSelectionResult::unwrap(): receiver ID {} was closed!",
                    id
                )
            },
        }
    }
}

#[derive(PartialEq, Debug)]
pub struct OsOpaqueIpcChannel {
    fd: c_int,
}

impl Drop for OsOpaqueIpcChannel {
    fn drop(&mut self) {
        // Make sure we don't leak!
        //
        // The `OsOpaqueIpcChannel` objects should always be used,
        // i.e. converted with `to_sender()` or `to_receiver()` --
        // so the value should already be unset before the object gets dropped.
        debug_assert!(self.fd == -1);
    }
}

impl OsOpaqueIpcChannel {
    pub(super) fn from_fd(fd: c_int) -> OsOpaqueIpcChannel {
        OsOpaqueIpcChannel { fd }
    }

    pub fn to_sender(&mut self) -> OsIpcSender {
        OsIpcSender::from_fd(mem::replace(&mut self.fd, -1))
    }

    pub fn to_receiver(&mut self) -> OsIpcReceiver {
        OsIpcReceiver::from_fd(mem::replace(&mut self.fd, -1))
    }
}

pub struct OsIpcOneShotServer {
    fd: c_int,

    // Object representing the temporary directory the socket was created in.
    // The directory is automatically deleted (along with the socket inside it)
    // when this field is dropped.
    _temp_dir: tempfile::TempDir,
}

impl Drop for OsIpcOneShotServer {
    fn drop(&mut self) {
        unsafe {
            let result = libc::close(self.fd);
            assert!(thread::panicking() || result == 0);
        }
    }
}

impl OsIpcOneShotServer {
    pub fn new() -> Result<(OsIpcOneShotServer, String), UnixError> {
        unsafe {
            let fd = libc::socket(libc::AF_UNIX, libc::SOCK_SEQPACKET | utils::SOCK_FLAGS, 0);
            let temp_dir = tempfile::Builder::new().tempdir().unwrap();
            let socket_path = temp_dir.path().join("socket");
            let path_string = socket_path.to_str().unwrap();

            let path_c_string = CString::new(path_string).unwrap();
            let (sockaddr, len) = utils::new_sockaddr_un(path_c_string.as_ptr());
            if libc::bind(
                fd,
                &sockaddr as *const _ as *const libc::sockaddr,
                len as libc::socklen_t,
            ) != 0
            {
                return Err(UnixError::last());
            }

            if libc::listen(fd, 10) != 0 {
                return Err(UnixError::last());
            }

            Ok((
                OsIpcOneShotServer {
                    fd,
                    _temp_dir: temp_dir,
                },
                path_string.to_string(),
            ))
        }
    }

    #[allow(clippy::type_complexity)]
    pub fn accept(
        self,
    ) -> Result<
        (
            OsIpcReceiver,
            Vec<u8>,
            Vec<OsOpaqueIpcChannel>,
            Vec<OsIpcSharedMemory>,
        ),
        UnixError,
    > {
        unsafe {
            let sockaddr: *mut libc::sockaddr = ptr::null_mut();
            let sockaddr_len: *mut libc::socklen_t = ptr::null_mut();
            let client_fd = libc::accept(self.fd, sockaddr, sockaddr_len);
            if client_fd < 0 {
                return Err(UnixError::last());
            }
            utils::make_socket_lingering(client_fd)?;

            let receiver = OsIpcReceiver::from_fd(client_fd);
            let (data, channels, shared_memory_regions) = receiver.recv()?;
            Ok((receiver, data, channels, shared_memory_regions))
        }
    }
}
